package com.etlmaster.util.vfs;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.commons.vfs2.FileObject;
import org.apache.commons.vfs2.FileType;
import org.apache.commons.vfs2.provider.AbstractFileName;
import org.apache.commons.vfs2.provider.AbstractFileObject;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;



/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HdfsFileObject extends AbstractFileObject {

	private FileSystem fileSystem;
	
	private final static Logger LOG = LoggerFactory.getLogger(HdfsFileObject.class);

	protected HdfsFileObject(AbstractFileName name, HdfsFileSystem fs) {
		super(name, fs);
		this.fileSystem = fs.getHdfsFileSystem();
	}

	protected FileType doGetType() throws Exception {
		FileStatus fileStatus = null;
		try {
			fileStatus = this.fileSystem.getFileStatus(new Path(getName().getPath()));
		} catch (FileNotFoundException e) {
			LOG.warn("获取文件类型出错{}",e.getMessage());;
		}
		if (fileStatus == null)
			return FileType.IMAGINARY;
		if (fileStatus.isDirectory()) {
			return FileType.FOLDER;
		}
		return FileType.FILE;
	}

	protected String[] doListChildren() throws Exception {
		FileStatus[] fileStatus = this.fileSystem.listStatus(new Path(getName().getPath()));
		String[] children = new String[fileStatus.length];
		for (int i = 0; i < fileStatus.length; i++) {
			children[i] = fileStatus[i].getPath().getName();
		}
		return children;
	}

	protected long doGetContentSize() throws Exception {
		Path path = new Path(getName().getPath());
		return this.fileSystem.getContentSummary(path).getLength();
	}

	protected InputStream doGetInputStream() throws Exception {
		return this.fileSystem.open(new Path(getName().getPath()));
	}

	protected void doDelete() throws Exception {
		try {
			this.fileSystem.delete(new Path(getName().getPath()), true);
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	protected void doRename(FileObject newfile) throws Exception {
		Path src = new Path(getName().getPath());
		Path dst = new Path(newfile.getName().getPath());
		this.fileSystem.rename(src, dst);
	}

	protected void doCreateFolder() throws Exception {
		try {
			this.fileSystem.mkdirs(new Path(getName().getPath()));
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	protected OutputStream doGetOutputStream(boolean bAppend) throws Exception {
		Path path = new Path(getName().getPath());
		if (bAppend && fileSystem.exists(path)) {
			return this.fileSystem.append(path);
		}
		return this.fileSystem.create(path);
	}
}